package com.gugee.ins.data.collector.comment.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.beanutil.InsCommentConversion;
import com.gugee.ins.common.constant.AiConst;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.comment.InsCommentDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsRelBloggerActiveFollowerDao;
import com.gugee.ins.common.dal.service.media.InsMediaDao;
import com.gugee.ins.common.dal.spider.InsMediaCommentSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsBloggerMain;
import com.gugee.ins.common.model.mysql.comment.InsComment;
import com.gugee.ins.common.model.mysql.media.InsMedia;
import com.gugee.ins.common.model.spider.blogger.PageInfo;
import com.gugee.ins.common.model.spider.comment.EdgeMediaCommentBody;

import com.gugee.ins.data.collector.comment.config.InsCommentConfig;
import com.gugee.ins.data.collector.comment.service.InsCommentService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsCommentServiceImpl implements InsCommentService {

    /**
     * 配置
     */
    @Autowired
    InsCommentConfig insCommentConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * 爬虫接口
     */
    @Autowired
    InsMediaCommentSpiderClient insMediaCommentSpiderClient;

    /**
     * InsCommentDao
     */
    @Autowired
    InsCommentDao insCommentDao;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * InsBloggerMainDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;

    /**
     * InsMediaDao
     */
    @Autowired
    InsMediaDao insMediaDao;

    /**
     * InsRelBloggerActiveFollowerDao
     */
    @Autowired
    InsRelBloggerActiveFollowerDao insRelBloggerActiveFollowerDao;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insCommentCollect-worker";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insCommentConfig.getInsCommentCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理InsComment收入任务
     * @param insBloggerMessages
     */
    @Override
    public void handleInsCommentCollect(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    //采集播主关注人数据
                    runCollectInsComment(msg);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    log.error("InsComment collect error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("InsComment collect  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集更新评论数据
     * @param insBloggerMessage
     */
    public void runCollectInsComment(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessage.getUid();
        List<InsComment> insComments;
        // 评论id，去除重复的评论
        Set<Long> cids;
        // 获取近30天的帖子
        long timeRang = start / 1000 - insCommentConfig.getLatestDays() * 24 * 3600;
        List<InsMedia> insMedias = insMediaDao.findByOwnerUidAndTakenTime(uid,timeRang, insCommentConfig.getInsCommentCollectMediaMinCount());
        if(!insMedias.isEmpty()){
            PageInfo pageInfo;

            // 已采集的评论数
            int collectCommentCount = 0;
            // 每个帖子预估采集评论数
            int eachMediaCollectComment = insCommentConfig.getInsCommentCollectCount() / insMedias.size()
                    + insCommentConfig.getInsCommentCollectBaseCount();

            // todo insComments采集回来的评论可能会重复？
            for(InsMedia insMedia : insMedias){
                insComments = new LinkedList<>();
                cids = new HashSet<>();
                pageInfo = spiderComment(insMedia.getShortcode(),insCommentConfig.getInsSpiderCommentEachCount(),"",insComments,uid,cids);
                while(pageInfo != null && pageInfo.getHasNextPage()
                        && !StringUtils.isEmpty(pageInfo.getEndCursor())
                        && eachMediaCollectComment > insComments.size()){
                    pageInfo = spiderComment(insMedia.getShortcode(),insCommentConfig.getInsSpiderCommentEachCount(),pageInfo.getEndCursor(),insComments,uid,cids);
                }
                if(!insComments.isEmpty()){
                    collectCommentCount += insComments.size();
                    // 保存帖子评论信息
                    saveComment(insMedia.getShortcode(),insComments);
                }
                if(collectCommentCount >= insCommentConfig.getInsCommentCollectCount()){
                    break;
                }
            }

            if(collectCommentCount > 0){
                // 更新博主与活跃粉丝关系
                try{
                    // 插入不存在的
                    insRelBloggerActiveFollowerDao.batchInsertByUid(uid,insCommentConfig.getInsCommentCollectActiveFollowerCount());
                }catch (Exception e){
                    log.error("update ins Rel Blogger Active Follower error.msg : {}" ,e.getMessage());
                }
                // 更新标记位
                insBloggerMainDao.updateCommentCollectTime(uid,collectCommentCount);

                // 发送头像、语言分析、活跃粉丝关注、地理位置消息
                sendPriorityMsg(insBloggerMessage);
            }
        }

        if ((System.currentTimeMillis() - start) / 1000 > insCommentConfig.getCommentDebugTimeLimit()) {
            log.info("InsMedia collect uid:{} use time:{}s", insBloggerMessage.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 循环拉评论
     * @param shortcode
     * @param count
     * @param endCursor
     * @param cids
     * @return
     */
    private PageInfo spiderComment(String shortcode,int count,String endCursor,List<InsComment> insComments,long uid,Set<Long> cids){
        //根据uid爬取信息
        List<EdgeMediaCommentBody.EdgeMediaComment> edges;
        EdgeMediaCommentBody edgeMediaCommentBody;
        PageInfo pageInfo;
        long time = System.currentTimeMillis() / 1000;
        try {
            edgeMediaCommentBody = insMediaCommentSpiderClient.getMediaComments(shortcode,count,endCursor);
            edges = edgeMediaCommentBody.getEdges();
            if(edges != null && !edges.isEmpty()){
                for(EdgeMediaCommentBody.EdgeMediaComment edge : edges){
                    collectComments(edge,insComments,shortcode,uid,cids,time);
                }
            }
            pageInfo = edgeMediaCommentBody.getPageInfo();
        }catch (Exception e){
            pageInfo = null;
            log.error("spider ins comment collect error, msg :{}",e.getMessage());
        }
        return pageInfo;
    }

    /**
     *
     * @param edge
     * @param insComments
     * @param shortcode
     * @param uid
     * @param cids
     * @param time
     */
    public void collectComments(EdgeMediaCommentBody.EdgeMediaComment edge,List<InsComment> insComments,String shortcode,long uid,
                    Set<Long> cids,long time){
        try{
            if(edge != null && edge.getNode() != null){
                InsComment insComment = InsCommentConversion.mapper(edge.getNode());
                if(insComment.getCUid() != uid
                        && insComment.getId() != null
                        && !cids.contains(insComment.getId())){
                    insComment.setMUid(uid);
                    insComment.setShortcode(shortcode);
                    insComment.setCreateTime(time);
                    insComments.add(insComment);
                    // 采集评论用户
                    if(insComment.getCUid() > 0){
                        sendPriorityCollectBloggerMsg(insComment.getCUid(),edge.getNode().getOwner().getUsername()
                                ,edge.getNode().getOwner().getPicUrl(),edge.getNode().getOwner().getIsPrivate(),
                                InsBloggerMsgTypeCodeEnum.CLP_SRC_ACTIVE_FOLLOWER.getCode());
                    }
                    cids.add(edge.getNode().getId());
                }
            }
        }catch (Exception ex){
            log.debug("ins comment id trans error, msg :{}",ex.getMessage());
        }
    }

    /**
     * 发送博主优先收录的消息
     * @param uid
     * @param userName
     * @param picUrl
     */
    public void sendPriorityCollectBloggerMsg(long uid,String userName,String picUrl,Boolean isVerified,int srcTypeCode){
        if(uid > 0 && insBloggerDao.checkCollecting(uid,true)){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName(userName);
            insBloggerMessage.setPicUrl(StringUtils.isEmpty(picUrl) ? "" : picUrl);
            insBloggerMessage.setRanking(isVerified != null ? isVerified : false);
            insBloggerMessage.setRetryLimitTimes(insCommentConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setRetryTimes(insCommentConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setTypeCode(srcTypeCode);

            kafkaTemplateInsBlogger.send(insCommentConfig.getInsBloggerPriorityCollectOnlineTopic(),insBloggerMessage);

        }
    }

    /**
     * 发送优先消息
     * @param insBloggerMessage
     */
    public void sendPriorityMsg(InsBloggerMessage insBloggerMessage){
        InsBloggerMain insBlogger = insBloggerMainDao.findByUid(insBloggerMessage.getUid());
        if(insBlogger != null){
            // 发送分析头像消息
            sendPriorityAnalysisAvatarMsg(insBloggerMessage,insBlogger);
            insBloggerMessage.setPicUrl("");
            // 发送分析语言消息
            sendPriorityAnalysisLanguageMsg(insBloggerMessage,insBlogger);
            // 发送采集活跃粉丝关注的消息
//            sendPriorityActiveFollowerFollowingClMsg(insBloggerMessage);
            // 发送采集地理位置的消息
            sendPriorityLocationClMsg(insBloggerMessage,insBlogger);
        }
    }

    /**
     * 发送优先分析头像的消息
     * @param insBloggerMessage
     * @param insBlogger
     */
    public void sendPriorityAnalysisAvatarMsg(InsBloggerMessage insBloggerMessage,InsBloggerMain insBlogger){
        if(!insBlogger.getIsAnonymousPic()
                && (StringUtils.isEmpty(insBlogger.getRace()) || (insBlogger.getGender() != AiConst.FEMALE
                && insBlogger.getGender() != AiConst.MALE))){
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.HANDLE_AND_DO_OTHER_OPT.getCode());
        }else{
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode());
        }
        insBloggerMessage.setPicUrl(insBlogger.getPicUrl());
        insBloggerMessage.setRetryTimes(insCommentConfig.getInsAnalysisAvatarPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insCommentConfig.getInsAnalysisAvatarPriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insCommentConfig.getInsAnalysisAvatarPriorityTopic(),insBloggerMessage);
    }

    /**
     * 发送优先分析语言的消息
     * @param insBloggerMessage
     * @param insBlogger
     */
    public void sendPriorityAnalysisLanguageMsg(InsBloggerMessage insBloggerMessage,InsBloggerMain insBlogger){
        if(StringUtils.isEmpty(insBlogger.getLanguage()) || AiConst.LANGUAGE_UNKNOWN.equals(insBlogger.getLanguage())){
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.HANDLE_AND_DO_OTHER_OPT.getCode());
        }else{
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode());
        }
        insBloggerMessage.setRetryTimes(insCommentConfig.getInsAnalysisLanguagePriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insCommentConfig.getInsAnalysisLanguagePriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insCommentConfig.getInsAnalysisLanguagePriorityTopic(),insBloggerMessage);
    }

    /**
     * 发送优先采集活跃粉丝的消息
     * @param insBloggerMessage
     */
    public void sendPriorityActiveFollowerFollowingClMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode());
        insBloggerMessage.setRetryTimes(insCommentConfig.getInsBloggerFollowingPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insCommentConfig.getInsBloggerFollowingPriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insCommentConfig.getInsBloggerFollowingPriorityTopic(),insBloggerMessage);
    }

    /**
     * 发送优先采地理位置的消息
     * @param insBloggerMessage
     * @param insBlogger
     */
    public void sendPriorityLocationClMsg(InsBloggerMessage insBloggerMessage,InsBloggerMain insBlogger){
        if(StringUtils.isEmpty(insBlogger.getCityName())
                && StringUtils.isEmpty(insBlogger.getDetailCityName())
                && !insBlogger.getIsPrivate()){
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.HANDLE_AND_DO_OTHER_OPT.getCode());
        }else{
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode());
        }
        insBloggerMessage.setRetryTimes(insCommentConfig.getInsBloggerLocationPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insCommentConfig.getInsBloggerLocationPriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insCommentConfig.getInsBloggerLocationPriorityTopic(),insBloggerMessage);
    }

    /**
     * 保存帖子评论信息
     * @param shortcode
     * @param insComments
     */
    @Transactional(rollbackFor = Exception.class)
    public void saveComment(String shortcode,List<InsComment> insComments){
        try{
            // 批量插入或更新 insComments
            insCommentDao.saveOrUpdateBatch(insComments);
            // 更新帖子采集评论的信息
            insMediaDao.updateMediaCommentInfo(shortcode,insComments.size());
        }catch (Exception e){
            log.error("insert / update ins comment error.msg : {}" ,e.getMessage());
        }
    }
}
